package com.atguigu.chapter08;

import com.atguigu.bean.LoginEvent;
import com.atguigu.util.MyUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/16 15:08
 */
public class Flink04_High_Login {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 创建WatermarkStrategy
        WatermarkStrategy<LoginEvent> wms = WatermarkStrategy
            .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20))
            .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                @Override
                public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                    return element.getEventTime();
                }
            });
        env
            .readTextFile("input/LoginLog.csv")
            .map(line -> {
                String[] data = line.split(",");
                return new LoginEvent(Long.valueOf(data[0]),
                                      data[1],
                                      data[2],
                                      Long.parseLong(data[3]) * 1000L);
            })
            .assignTimestampsAndWatermarks(wms)
            .keyBy(LoginEvent::getUserId)
            .process(new KeyedProcessFunction<Long, LoginEvent, String>() {
                
                private ListState<Long> failTsState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    failTsState = getRuntimeContext().getListState(new ListStateDescriptor<Long>("failTsState", Long.class));
                }
                
                @Override
                public void processElement(LoginEvent event,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    /*
                     使用列表状态, 存储失败的时间戳, 判断这两个时间戳是否在2s之内, 如果是则是恶意登录
                     
                    否则, 把第一个删除
                    
                    如果碰到成功, 则情况状态
                     */
                    if ("fail".equals(event.getEventType())) {
                        failTsState.add(event.getEventTime());
    
                        List<Long> list = MyUtil.toList(failTsState.get());
                        if (list.size() == 2) { // 如果有两个失败, 则按照时间戳进行升序排列
                            list.sort(Long::compareTo);
                            if ((list.get(1) - list.get(0)) <= 2000) {
                                out.collect(event.getUserId() + " 正在恶意登录");
                            }
                            failTsState.clear();
                            failTsState.add(list.get(1));
                        }
    
                    } else if ("success".equals(event.getEventType())) {
                        failTsState.clear();
                    }
                    
                }
            })
            .print();
        
        env.execute();
        
    }
}
